Rewrite async methods to return owned futures. - #149

See issue datrs#148

Add tests for replication
Fix bugs getting pending_missing_nodes

This raised questons of how things like event interfaces should be handled. If there are multiple subscribers to core events, then the events must be cloned, or only one person can get them.
Instead the user of the hypercore can decide how it is handled.

Note that some feature combos are still missing. mostly with 'sparse'
and 'cache' combos. I don't think this matters enough at this time to
add the extra test overhead

Lets a caller (e.g. corestore) drive one core's replication over a
Channel already opened on a Protocol it owns itself, instead of
Hypercore::replicate's self-owned Protocol/stream. Also switches the
hypercore-protocol dependency to a local path so its type matches
whatever corestore uses, and fixes an unrelated doctest broken by the
earlier append/append_batch &mut self -> &self change.

`Hypercore::update(UpdateOptions { wait, force, timeout })` resolves `Ok(true)`
when the core's verified length grew, `Ok(false)` when peers confirmed there is
nothing newer, and `Err(HypercoreError::Timeout)` when nobody answered in time.
Previously a reader could only discover new data by polling `info().length` or
by `get()`ing an index and hoping replication had reached it.

Two supporting changes were needed:

1. A peer registry. Each `ChannelState` now owns an `Arc<PeerSyncState>` of
   atomics (remote_synced / remote_length / remote_fork / remote_can_upgrade /
   length_acked / upgrade_inflight) and self-registers a `Weak` of it into
   `HypercoreInner::peers`. This lets the core see every peer's advertised
   length regardless of who drives replication — `Replicator`,
   `attach_replicator`, or corestore's `attach_channel` — which is what makes
   an honest `Ok(false)` possible. It is a port of JS's
   `_checkUpgradeIfAvailable` (replicator.js:2244). `Weak` means there is no
   deregistration to get wrong: entries are pruned on read.

2. Upgrade requests are retried. The request used to be built only inside
   `on_synchronize`, so a single misfire of its `length_acked == info.length`
   guard meant it was never sent at all. It is now
   `ChannelState::maybe_request_upgrade`, called from `on_synchronize` and once
   per poll, deduped on (fork, our length, their length) — Rust's `updateAll()`.

Two payload-free `Event` variants (`PeerSync`, `Upgrade`) carry the wake-ups.
The event channel is lossy (broadcast(32), overflow: true), so they are only
ever wake-ups; all authoritative state is read back from the registry.

Also adds `Hypercore::peer_count()`, a `HypercoreError::Timeout` variant, and a
plain tokio dependency for the timer. Drops a stray `dbg!()` from `GetFuture`.

Tested by 7 new cases in tests/replication.rs. Note `update` does not subsume
the existing `wait_for_length` helper: it guarantees verified length, not that
blocks have been downloaded (contiguous_length), and swapping it in fails
deterministically.

Known test gaps, to be addressed next: the per-poll retry in (2) is not covered
(removing it leaves all tests passing), and every test uses exactly one peer, so
the multi-peer case the registry exists for is unexercised, as is peer teardown.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Adds multi-peer coverage for `Hypercore::update`, and refactors the decision it
rests on so that coverage is worth something.

The obvious integration tests turned out not to test anything. Two peers
attached to one reader, 60 consecutive green runs — and then mutating the peer
loop to `peers.iter().take(1)`, the "first responder wins" design the registry
exists to avoid, left all 17 integration tests passing. In the all-in-sync case
both peers are useless so checking one or all gives the same answer, and in the
mixed case the retry loop masked the difference, because `maybe_request_upgrade`
upgrades the reader regardless of what the predicate decides.

So `any_peer_may_upgrade` moves out of `impl UpdateFuture` into a free function
over `(&[Arc<PeerSyncState>], &Info, &Snapshot, wait)` and is unit tested
directly, with 8 cases in src/replication/update.rs. Multi-peer states are a
race to stage over real connections and exact to construct by hand. The
multi-peer case asserts both peer orderings so it cannot pass by luck. Five
mutations were used to confirm the tests bite: consulting only peers[0] (kills
2), a useless peer ending the search (3), dropping the fork check (1), never
waiting for an unsynced peer (1), and ignoring upgrade_inflight (1).

The integration test `update_with_two_peers_only_one_useful` also asserted
something the design does not guarantee. A probe showed a single useless peer
makes `update` return Ok(false) in ~58ms rather than holding out for peers that
have not connected — correct, and matching JS's `_checkUpgradeIfAvailable`,
which likewise iterates only the peers present at that instant. It passed on
timing luck; had the empty peer won the registration race the assert would have
failed. It now asserts the postcondition, with the reasoning in a comment.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

… bits

`Entry::decode` tested flag bit 2 (tree nodes) for the tree upgrade and
bitfield update as well, where `encode` writes bits 4 and 8. Local appends
set all three, so they decoded by accident. Replicated blocks store nodes
and a bitfield update without an upgrade, so reopening a core that had
received data from a peer failed with "Compact encoding failed, out of
bounds". Entries without tree nodes silently lost their upgrade and
bitfield update.

The bytes on disk were always correct, so existing stores open without
migration.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Also add some logs for debugging when replications goes wrong

A peer requests an upgrade only once we have acked its length, and the
ack travels in our Synchronize. `on_synchronize` only answered the first
one, so if the writer appended, and sent its data-upgrade Synchronizes,
before it learned the reader's new length, every one of them carried a
stale length and nothing ever sent a fresh one. The reader stalled one
upgrade behind for good. JS `onsync` calls `signalUpgrade()` whenever
its length is greater than the peer's; do the same.

Timing-dependent, so it only showed on a multi-threaded runtime (found
through hrss). The regression test polls both replicators by hand to
force the ordering, and fails without the fix.

Also records the two earlier oplog fixes in the CHANGELOG.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

`get` on a block not downloaded yet only waited when a replicator was
attached with `attach_replicator`. With a channel from `attach_channel`
(corestore's path) or a spawned `replicate`, it returned `Ok(None)` at
once, which hyperbee reported as `NoBlockAtSeqError` right after an
upgrade. It now also waits while the core's peer registry has a live
channel, and still returns `Ok(None)` straight away when nothing is
replicating.

It also subscribed to events after emitting `Get`, despite its comment;
with the channel driven by another task the reply could be missed.
Subscribe first.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>